package org.example.datastream.test.custom;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.example.util.Utils;

import java.math.BigDecimal;
import java.sql.*;
import java.util.List;


public class EnterpriseJdbcMultiSource extends RichSourceFunction<Enterprise> {

    private Connection[] conns;
    private PreparedStatement[][] pss;
    private ResultSet[][] rss;

    private String proxy;
    private String driver;
    private List<DbConf> dbs;
    private int tbSize;
    private String tb;
    private int dbSize;
    private int fetchSize;

    public EnterpriseJdbcMultiSource(String proxy, String driver, List<DbConf> dbs, String tb, int tbSize, int fetchSize) {
        this.proxy = proxy;
        this.driver = driver;
        this.dbs = dbs;
        this.tb = tb;
        this.tbSize = tbSize;
        this.dbSize = dbs.size();
        this.fetchSize = fetchSize;
        this.conns = new Connection[this.dbSize];
        this.pss = new PreparedStatement[this.dbSize][this.tbSize];
        this.rss = new ResultSet[this.dbSize][this.tbSize];
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        Class.forName(driver);

        for (int i = 0; i < dbSize; i ++) {
            DbConf db = dbs.get(i);
            String url = String.format("jdbc:%s//%s:%s/%s?useSSL=false&useUnicode=true&characterEncoding=UTF-8",
                    proxy, db.getHost(), db.getPort(), db.getDb());
            conns[i] = DriverManager.getConnection(url, db.getUsername(), db.getPassword());
            for (int j = 0; j < tbSize; j ++) {
                String sql = String.format("select * from %s_%d", tb, j);
                PreparedStatement ps = conns[i].prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
                ps.setFetchDirection(ResultSet.FETCH_REVERSE);
                ps.setFetchSize(fetchSize);
                pss[i][j] = ps;
            }
        }
    }

    @Override
    public void run(SourceContext<Enterprise> sourceContext) throws Exception {
        for (int i = 0; i < pss.length; i ++) {
            for (int j = 0; j < pss[i].length; j++) {
                ResultSet rs  = pss[i][j].executeQuery();

                while (rs.next()) {
                    Enterprise item = new Enterprise();
                    item.setEid(rs.getString("eid"));
                    item.setEnterpriseName(rs.getString("name"));
                    item.setStandardName(rs.getString("format_name"));
                    item.setGroupName(rs.getString("group_name"));
                    item.setGroupAbbreviation(rs.getString("group_abbreviation"));
                    item.setLogoUrl(rs.getString("logo_url"));
                    item.setOperName(rs.getString("oper_name"));
                    item.setOperType(rs.getString("oper_type"));
                    item.setOperNameId(rs.getString("oper_name_id"));
                    item.setTitle(rs.getString("title"));

                    item.setTitleCode(1);//rs.getInt("title_code") //todo:转换逻辑
                    item.setRecCapi(rs.getBigDecimal("regist_capi_new"));
                    String actualCapi = rs.getString("actual_capi");
                    BigDecimal actualCapiDecimal = Utils.getDecimalFromStr(actualCapi);
                    item.setRecCapi(actualCapiDecimal);
                    item.setCurrencyUnit(rs.getString("currency_unit"));
                    item.setBusinessUrl(rs.getString("url"));
                    item.setOperName(rs.getString("collegues_num"));
                    item.setBelongOrg(rs.getString("belong_org"));
                    item.setDepartment(rs.getString("department"));
                    item.setScope(rs.getString("scope"));
                    item.setRegNo(rs.getString("reg_no"));

                    item.setCreditNo(rs.getString("credit_no"));
                    item.setOrgNo(rs.getString("org_no"));
                    item.setEconKindCode(1);//rs.getInt("econ_kind_code") //todo:转换逻辑
                    item.setType(rs.getInt("type"));
                    item.setTypeNew(rs.getString("type_new"));
                    item.setCategoryNew(rs.getString("category_new"));
                    item.setStatus(rs.getString("status"));
                    item.setStatusCode(1);//rs.getInt("new_status_code") //todo:转换逻辑
                    Date startDate = Utils.str2Date2(rs.getString("start_date"));
                    item.setStartDate(startDate);
                    Date termStart = Utils.str2Date2(rs.getString("term_start"));
                    item.setTermStart(termStart);

                    Date termEnd = Utils.str2Date2(rs.getString("term_end"));
                    item.setTermEnd(termEnd);
                    Date checkDate = Utils.str2Date2(rs.getString("check_date"));
                    item.setCheckDate(checkDate);
                    Date logoutDate = Utils.str2Date2(rs.getString("logout_date"));
                    item.setLogoutDate(logoutDate);
                    item.setLogoutReason(rs.getString("logout_reason"));
                    Date revokeDate = Utils.str2Date2(rs.getString("revoke_date"));
                    item.setRevokeDate(revokeDate);
                    item.setRevokeReason(rs.getString("revoke_reason"));
                    item.setRevokedCertificates(rs.getString("revoked_certificates"));
                    item.setProvinceCode(rs.getString("province_code"));
                    item.setSource(rs.getString("source"));
                    item.setRowUpdateTime(rs.getTimestamp("row_update_time"));

//            long createdTimeNum = rs.getLong("created_time");
//            LocalDateTime dateTime = Utils.timestamp2Datetime(createdTimeNum);
//            Timestamp timestamp = Timestamp.valueOf(dateTime);
                    item.setCreateTime(rs.getTimestamp("row_update_time"));

                    sourceContext.collect(item);
                }
            }
        }
    }

    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        super.close();

        for (Connection conn : conns) {
            if (null != conn) {
                conn.close();
            }
        }
        for (PreparedStatement[] psa : pss) {
            for (PreparedStatement ps : psa) {
                if (null != ps) {
                    ps.close();
                }
            }
        }
    }
}
